package hyl.base.mq.rocket;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.List;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import hyl.base.mq.BaseMQ;
import hyl.core.MyCharset;
import hyl.core.MyFun;
import hyl.core.info.CMsg;

/**
 * 用法 : 发送 :<br>
 * 实例化 iniProducer() --> 准备 ready() -->发送 send() --> 关闭 shutdownProducer()<br>
 * 接收:<br>
 * 实例化 iniConsumer() --> 准备 ready() -->接收 receive :包含开启 startConsume()--> 关闭
 * shutdownProducer()<br>
 * 
 * @author 37798955@qq.com
 *
 */
public class MyRocketMq extends BaseMQ {
	DefaultMQProducer producer;
	DefaultMQPushConsumer consumer;// 系统自主控制的消费者
	// MQPullConsumer 使用者自己控制的消费者
	// MQConsumer 是接口
	// 配置 读取mq文件夹中的rocketmq配置
	RktConfig RC;

	protected MyRocketMq(String 配置项, String group) {
		if (!MyFun.isEmpty(配置项))
			try {
				RC = RktConfig.getInstance(配置项, group);
			} catch (IOException e) {
				e.printStackTrace();
			}
	}

	/**
	 * 实例化生产者,专门用来发消息
	 * 
	 * @param 配置项
	 * @param group
	 * @return
	 */
	public static MyRocketMq iniProducer(String 配置项, String group) {
		return iniProducer(配置项, group, 0);

	}

	/**
	 * 实例化生产者,专门用来发消息
	 */
	public static MyRocketMq iniProducer(String 配置项, String group, int clientid) {
		MyRocketMq mq = new MyRocketMq(配置项, group);

		mq.producer = new DefaultMQProducer(mq.RC.getGroup());
		// mq.producer.setProducerGroup(mq.RC.getGroup());
		// 是否启用vip netty通道以发送消息
		mq.producer.setVipChannelEnabled(false);
		mq.producer.setNamesrvAddr(mq.RC.getNameServer());
		// 发送失败以后重试次数
		mq.producer.setRetryTimesWhenSendFailed(mq.RC.getRetryTimeFailed());
		// 发送别名
		mq.producer.setUnitName("p_" + String.valueOf(clientid));
		return mq;
	}

	public void startProducer() throws MQClientException {
		producer.start();
	}

	/**
	 * 实例化消费者 专门用来收消息的
	 * 
	 * @param 配置项
	 * @param group
	 * @param clientid
	 * @return
	 */
	public static MyRocketMq iniConsumer(String 配置项, String group, int clientid) {
		return iniConsumer(配置项, group, clientid, 0);
	}

	public static MyRocketMq iniConsumer(String 配置项, String group, int clientid, int 每批拉取) {// ,int id
		MyRocketMq mq = new MyRocketMq(配置项, group);
		mq.consumer = new DefaultMQPushConsumer(mq.RC.getGroup());
		// 消费组的名称，用于标识一类消费者
		// mq.consumer.setConsumerGroup(mq.RC.getGroup());
		// 是否启用vip netty通道以发送消息
		mq.consumer.setVipChannelEnabled(false);
		/* 开启广播模式,推送端不用变,订阅端改变配置 */
		mq.consumer.setNamesrvAddr(mq.RC.getNameServer());
		// 消费模式:MessageModel.CLUSTERING 集群消费模式
		// MessageModel.BROADCASTING 广播消费模式
		mq.consumer.setMessageModel(mq.RC.get消费模式());
		// mq.consumer.setLanguage(LanguageCode.JAVA);
		mq.consumer.setUnitName("c_" + String.valueOf(clientid));
		if (每批拉取 > 0)
			mq.consumer.setPullBatchSize(每批拉取);
		else
			mq.consumer.setPullBatchSize(mq.RC.get每批最多拉取());
		// mq.consumer.setPullInterval(mq.RC.get拉取间隔());//设置消息拉取间隔，默认为0
		// mq.consumer.setInstanceName(instanceName);
		// 设置负载均衡算法实现策略 默认值：AllocateMessageQueueAveragely（取模平均分配）
		// mq.consumer.setAllocateMessageQueueStrategy(allocateMessageQueueStrategy);
		// Consumer启动后，默认从什么位置开发消费，默认为CONSUME_FROM_LAST_OFFSET
		// mq.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

		// consumer.setSubscription()：//不建议设置，订阅topic建议直接调用subscribe接口设置订阅关系
		// consumer.setOffsetStore():设置消息进度存储

		mq.consumer.setConsumeThreadMax(mq.RC.get最多线程());// 设置消费线程的最大数量，默认为64个线程

		mq.consumer.setConsumeThreadMin(mq.RC.get最少线程());// 设置消费线程的最小数量，默认为20个线程
		// 本地队列缓存消息最大数，默认为1000
		mq.consumer.setPullThresholdForQueue(mq.RC.get缓存队列消息());
		// 批量消费消息，一次消费多少条消息，默认为1条不设置
		// consumer.setConsumeMessageBatchMaxSize()：

		/*
		 * consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		 * CONSUME_FROM_LAST_OFFSET 默认策略， 从该队列最尾开始消费，即跳过历史消息<br>
		 * CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费， 即历史消息（还储存在broker的）全部消费一遍<br>
		 * CONSUME_FROM_TIMESTAMP从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前<br>
		 */
		mq.consumer.setConsumeFromWhere(mq.RC.getConsumeFrom());
		// mq.consumer.setConsumeMessageBatchMaxSize(mq.RC.getConsumeBatchMaxSize());
		return mq;
	}

	// 发送完消息之后，调用shutdown()方法关闭producer
	public void shutdownProducer() {
		producer.shutdown();
	}

	public void shutdownConsume() {
		consumer.shutdown();
	}

	/**
	 * 阻塞
	 */
	public void suspend() {
		consumer.suspend();
	}

/////////////////////////////定义监听器////////////////////

	/**
	 * 收到以后按照新的格式封装成CMsg后,再处理
	 * 
	 * @param msg
	 * 
	 */
	void receivemsg(MessageExt msg) {
		// System.out.println(msg.toString());
		// System.out.println(msg);
		CMsg cm = new CMsg();
		cm.s主题 = msg.getTopic();
		cm.s键 = msg.getTags();
		// msg.getKeys();
		cm.bs内容 = msg.getBody();

		// MyFun.print("receivemsg()",cm.s主题,cm.s键,cm.getS内容());
		if (_callback != null)
			try {
				_callback.success(cm);
			} catch (Exception e) {
				_callback.error(cm, e); // 1s 2s 5s ... 2h
			}

		// 调试语句 ....
//		String msgBody = null;
//		try {
//			msgBody = new String(cm._ba内容, "utf-8");
//		} catch (UnsupportedEncodingException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
//
//		System.out.println("收到消息： topic:" + msg.getTopic() + " ,tags:" + msg.getTags() + " ,key:" + msg.getKeys()
//				+ " ,msg: " + msgBody);
	}

	/**
	 * 设置消息监听器
	 * 
	 * @param listener
	 */

	void setConsumerListener(MessageListener listener) {

		if (listener == null) {
			if (_mode == 0) {
				// 并发消费 默认异步消费监听器
				listener = new MessageListenerConcurrently() {
					@Override
					public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
							ConsumeConcurrentlyContext context) {
						// ConsumeConcurrentlyContext 返回消费状态
						// CONSUME_SUCCESS 消费成功
						// RECONSUME_LATER 消费失败，需要稍后重新消费,重试
						// try {
						// 06接收消息并打印
						// System.out.println("异步");
						for (MessageExt msg : msgs) {
							receivemsg(msg);
						}
						// return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
				};
			} else {
				// 顺序消费 默认同步消费监听器
				listener = new MessageListenerOrderly() {
					@Override
					public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
						// context.setAutoCommit(false);
						// System.out.println("同步");
						for (MessageExt msg : msgs) {
							receivemsg(msg);
						}
						// return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
						return ConsumeOrderlyStatus.SUCCESS;
					}

				};
			}
		}

		consumer.setMessageListener(listener);
	}

/////////////////////////////定义底层封装API////////////////////
	public Message createMsg(String topic, String key, String value) {
		byte[] bys;
		try {
			bys = value.getBytes(RemotingHelper.DEFAULT_CHARSET);
		} catch (UnsupportedEncodingException e) {
			return null;
		}
		Message msg = new Message(topic, key, bys); /* Message body */
		return msg;
	}

	public Message createMsg(String topic, String key, byte[] value) {
		Message msg = new Message(topic, key, value);/* Message body */

		return msg;
	}

	/**
	 * 延后发送 延迟级别 与 时间 1 1s,2 5s,3 10s,4 30s,5 1m,6 2m ...,14 10m,15 20m,16 30m,17
	 * 1h,18 2h
	 */
	public Message createMsg(String topic, String key, String value, Integer delayTimeLevel)
			throws UnsupportedEncodingException {
		Message msg = createMsg(topic, key, value);
		if (delayTimeLevel == null)
			delayTimeLevel = RC.getDelayTimeLevel();
		msg.setDelayTimeLevel(delayTimeLevel);// 3level= 10s
		return msg;
	}

	/**
	 * 应用:异步并行发送,回调函数确认并处理返回结果
	 * 
	 * 
	 */
	public void sendWithCall(Message msg, SendCallback call)
			throws MQClientException, RemotingException, InterruptedException {
		producer.send(msg, call);
	}

	/**
	 * 
	 * 应用:适用于日志
	 * 
	 */
	public void sendLog(Message msg) throws MQClientException, RemotingException, InterruptedException {
		producer.sendOneway(msg);
	}

	/**
	 * 应用:并行发送
	 * 
	 * 
	 */
	public SendResult sendOne(Message msg)
			throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		return producer.send(msg);

	}

	MessageQueueSelector _selector = null;

	/**
	 * 应用:并行发送(带消息序号)
	 * 
	 * MessageQueueSelector 可以参考本类用例
	 */
	public SendResult sendWithID(int orderId, Message msg, MessageQueueSelector selector)
			throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		if (msg == null)
			return null;
		// 如果 有新的select 用新的,如果没有用旧的,如果旧的也没有就创建一个
		if (selector == null) {
			if (_selector == null) {
				// 通道选择 同步发送
				_selector = new MessageQueueSelector() {
					@Override
					public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
						return mqs.get(0);
					}
				};
			}
		} else {
			_selector = selector;
		}
		return producer.send(msg, _selector, orderId);

	}

	///////////////////// 对外简化接口--最常用的功能///////////
	@Override
	public void send(String 小标题, byte[] msg) {
		send(_topic, 小标题, msg);
	}
	public void send(String tag, String msg) {
		send(tag,msg.getBytes(MyCharset._UTF8));
	}
	public void send(String topic, String tag, String msg) {
		send(topic,tag,msg.getBytes(MyCharset._UTF8));
	}
	public void send(String topic, String tag, byte[] msg) {
		Message msg1 = createMsg(topic, tag, msg);
		CMsg msg2 = new CMsg();
		msg2.s主题 = topic;
		msg2.s键 = tag;
		msg2.bs内容 = msg;

		try {
			// System.out.println(msg2.toJsonString());
			sendOne(msg1);
			if (_callback != null)
				_callback.success(msg2);
		} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
			if (_callback != null)
				_callback.error(msg2, e);
		}
	}

	// String _tags="*";
	/**
	 * 消息订阅 在生产环境不需要 回滚,可以人工处理, 所以建议直接自动提交,不需要手动提交
	 * <p>
	 * Title: startConsume
	 * </p>
	 * <p>
	 * Description: 一定要先ready 才能用receive 否则会无法获得返回值
	 * </p>
	 */

	@Override
	public void receive() {
		try {

//			for(java.util.Map.Entry<String, Set<String>> item:_topics.entrySet()) {
//				String keys=MyFun.join(item.getValue().toArray(new String[0] ), " || ")	;
//				keys=MyFun.nvlStr(keys, "*");
//				consumer.subscribe(item.getKey(), keys);
//			}
			//MyFun.print(_topic, "  ", _key);
			consumer.subscribe(_topic, _key);

			// System.out.println("receive"+_topic);
			// consumer.subscribe(topic, messageSelector);
			// 已经有监听器了就不要改了,如果没有就创建一个
			if (null == consumer.getMessageListener())
				setConsumerListener(null);// 默认为异步接收监听

			consumer.start();
		} catch (MQClientException e) {
			if (_callback != null)
				_callback.error(null, e);
			e.printStackTrace();
		}
	}

	@Override
	public void close() {
		if (producer != null) {
			shutdownProducer();
			producer = null;
		} else if (consumer != null) {
			shutdownConsume();
			consumer = null;
		}
	}

}
